package org.common.utils.rpc.core;

import org.common.utils.net.netty.NettyClient;
import org.common.utils.rpc.entity.RpcRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;

/**
 * RPC发送器
 * @author zhouzhibing
 * @date 2025/11/19 10:59
 */
public class RpcSenders {

    private static final Logger log = LoggerFactory.getLogger(RpcSenders.class);
    /**
     * 连接对象
     */
    private final RpcConnection rpcConnection;
    /**
     * 请求池
     */
    private final RpcRequestPool requestPool;

    public RpcSenders(RpcConnection rpcConnection , RpcRequestPool requestPool) {
        this.rpcConnection = rpcConnection;
        this.requestPool = requestPool;
    }

    /**
     * 只发送请求，不接收响应结果。
     * @param request 请求参数
     */
    public void send(RpcRequest request) {
        try {
            Objects.requireNonNull(request);
            NettyClient client = getClient(request);
            client.sendMessage(request);
        } catch (Exception e) {
            log.error("rpc send request error, request : {}" , request , e);
        }
    }

    /**
     * 发送请求，并获取响应结果。
     * @param request 请求参数
     * @return 响应结果
     */
    public <T> CompletableFuture<T> sendFuture(RpcRequest request) {
        Objects.requireNonNull(request);
        CompletableFuture<T> future = new CompletableFuture<>();
        request.setFuture(future);

        try {
            NettyClient client = getClient(request);
            client.sendMessage(request);
            requestPool.addRequest(request);
        } catch (Exception e) {
            log.error("rpc send request error, request : {}" , request , e);
            future.completeExceptionally(e);
        }
        return future;
    }

    /**
     * 根据请求参数获取连接
     * @param request 请求参数
     * @return 连接对象
     */
    private NettyClient getClient(RpcRequest request) {
        Object o = request.getFirstArg();
        String key = null;
        //是基础类型参数，则使用参数值作为key
        if(o != null && isPrimitive(o)) {
            key = o.toString();
        } else {
            log.warn("rpc request param is not primitive type, not use hashConstant (注意，该方法第一个参数不是基础类型，没有使用到哈希定位算法。) , metadata = {} , arg = {}" , request.getMetadata() , o);
            key = System.currentTimeMillis() + "";
        }
        return rpcConnection.getClient(key);
    }


    /**
     * 判断对象是否为基本类型
     * @param o 对象
     * @return true/false
     */
    private boolean isPrimitive(Object o) {
        Class<?> aClass = o.getClass();
        String simpleName = aClass.getSimpleName();
        return switch (simpleName) {
            case "int", "Integer", "long", "Long", "short", "Short", "byte", "Byte",
                    "boolean", "Boolean", "float", "Float", "double", "Double", "String" -> { yield true; }
            default -> false;
        };
    }

}
